package cn.tpshion.redismq.handler;

import cn.tpshion.redismq.config.RedisMqProperties;
import cn.tpshion.redismq.data.UserMsg;
import com.alibaba.fastjson.JSON;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.stereotype.Component;
import org.yeauty.pojo.Session;

import java.io.BufferedReader;
import java.util.Objects;

@Slf4j
@Component
public class MessageListener implements InitializingBean, StreamListener<String, ObjectRecord<String, String>> {

    @Resource
    private StreamMessageListenerContainer<String, ObjectRecord<String, String>> container;

    @Resource
    private RedisMqProperties properties;

    @Resource
    private WebSocketHandler webSocketHandler;

    @Override
    public void afterPropertiesSet() {
        StreamOffset<String> streamOffset = StreamOffset.create(properties.getTopic(), ReadOffset.lastConsumed());

        StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder = StreamMessageListenerContainer.StreamReadRequest
                .builder(streamOffset)
                .errorHandler(new MessageErrorHandler())
                        .cancelOnError(throwable -> false);

        container.register(builder.build(), this);
    }

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        RecordId recordId = message.getId();
        String value = message.getValue();
        log.info("接收消息: recordId={} value={}", recordId, value);
        try {
            if (StringUtils.isBlank(value)) {
                return;
            }

            UserMsg userMsg = JSON.parseObject((String) JSON.parse(value), UserMsg.class);
            Session session = webSocketHandler.getSession(userMsg.getUserId());
            if (Objects.isNull(session)) {
                return;
            }

            // 发起请求
            RequestBody body = RequestBody.create(MediaType.get("application/json; charset=utf-8"),
                    "{\"bot_id\":\"7478216525708247078\",\"user_id\":\"123456\",\"stream\":true,\"additional_messages\":[{\"role\":\"user\",\"type\":\"question\",\"content_type\":\"text\",\"content\":\"你好\"}]}");

            // 构造请求
            Request request = new Request.Builder()
                    .url("https://api.coze.cn/v3/chat?conversation_id=7519696701909794856")
                    .post(body)
                    .addHeader("Content-Type", "application/json")
                    .addHeader("Authorization", "Bearer pat_Ey1d7Ix9urMKVE99xd7doplsMGAhE4UpwDqe7mP9F5mACb1mJB39q3zAf06p1UQn")
                    .build();

            OkHttpClient okHttpClient = new OkHttpClient();
            try (Response resp = okHttpClient.newCall(request).execute();
                 BufferedReader br = new BufferedReader(resp.body().charStream())) {

                String line;
                // 逐行读取
                while ((line = br.readLine()) != null) {
                    session.sendText(line);
                }

            }
        } catch (Exception e) {
            log.error("onMessage error:{}", e.getMessage(), e);
        }
    }
}
